package com.streamax.s17.mqs.business.mq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.streamax.s17.cloud.common.model.ResponseCode;
import com.streamax.s17.mqs.business.config.KafkaConfig;
import com.streamax.s17.mqs.business.config.InfluxDBConfig;
import com.streamax.s17.mqs.business.common.MqsBusinessException;
import com.streamax.s17.mqs.business.dao.entity.DeviceMessage;
import com.streamax.s17.mqs.business.service.IMsgQueryService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * @Description
 * @ClassName MessageConsumerI
 * @Author liwei
 * @Date 2019/8/14 10:42
 * @Version 2.0.2
 */
@Component
@Slf4j
public class MessageConsumer implements ApplicationRunner {

    @Autowired
    KafkaConfig kafkaConfig;

    @Autowired
    InfluxDBConfig dbConfig;

    @Autowired
    IMsgQueryService mqsService;

    public static final String groupId = "group.mqs";

    public MessageConsumer(){}


    @Override
    public void run(ApplicationArguments args) throws Exception {
        consume();
    }

    public void consume() {
        String broker = kafkaConfig.getBootstrapServers();
        String topics[] = kafkaConfig.getConsumer().getTopics();

        //设置消费组的名称
        //将属性值反序列化
        Properties properties = new Properties();
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("enable.auto.commit", true);
        properties.put("auto.commit.interval.ms", 1000);
        properties.put("auto.offset.reset", "earliest");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("bootstrap.servers", broker);
        properties.put("group.id", groupId);

        //创建一个消费者客户端实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        //订阅主题
        for(String topic : topics) {
            consumer.subscribe(Collections.singletonList(topic));
        }

        //循环消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            List<DeviceMessage> msgBeans = new ArrayList<>();
            for (ConsumerRecord<String, String> record : records) {
                log.debug("---[MQ]消费MQ消息:" + record.value());
                String val = record.value();
                DeviceMessage msg = convert2Bean(val);
                msgBeans.add(msg);
            }
            //TODO 持久化
            if (!msgBeans.isEmpty()){
                mqsService.persistenceMsg(msgBeans, dbConfig.getMeasurement());
            }
        }
    }
    private DeviceMessage convert2Bean(String value){
        if (!JSONUtil.isJson(value)){
            throw new MqsBusinessException(ResponseCode.BUSINESS_ERROR, "---[MQ]消息不是JSON：" + value);
        }
        JSONObject jsonObj = JSONUtil.parseObj(value);
        DeviceMessage tempMsg =  jsonObj.toBean(DeviceMessage.class);
        tempMsg.setOccurTimeVal(calcVal(tempMsg.getOccurTime()));
        return tempMsg;
    }

    private long calcVal(String time){
       String timeVal = time.replace("-", "").replace(":", "").replace(" ","");
       return Long.valueOf(timeVal);
    }
}

